package com.rem.flink.flink10Sql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 不从处理datastream流环境转换成处理table 的环境 直接 new 对象 处理
 * 可以直接从有界流/无界流 中拿到数据到 table 中
 * 将表对象的数据 不转换 通过输出表进行输出
 *
 * @author Rem
 * @date 2022-10-28
 */

public class TableEnvironmentTest {

    public static void main(String[] args) {
        /**
         * 2.1 基于blink进行流处理
         * 创建EnvironmentSettings的默认实例。
         * 在这种模式下，有界和无界数据流都可以被处理。
         */
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment tableEnv = TableEnvironment.create(settings);


        /**
         * 2.2 基于blink进行流处理
         * 批量创建EnvironmentSettings默认实例。
         * 此模式针对批处理场景进行了高度优化。只有有界数据流可以是 在此模式下处理。
         */
        EnvironmentSettings settings2 = EnvironmentSettings.inBatchMode();
        TableEnvironment tableEnv2 = TableEnvironment.create(settings2);


        /**
         * 3 创建虚拟表  用于输入 将input/clicks.csv 中数据输入到createDdl
         */
        String tableName = "clickTable";
        String createDdl = "CREATE TABLE " + tableName + " (" +
                " user_name STRING, " +
                " url STRING, " +
                " ts BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input/clicks.csv', " +
                " 'format' =  'csv' " +
                ")";
        tableEnv.executeSql(createDdl);

        /**
         * 4 转换表 将虚拟表转为 table 对象
         */
        Table clickTable = tableEnv.from(tableName);

        /**
         * 5.1 对表进行操作
         */
        Table resultTable = clickTable.select($("user_name"), $("url")).where($("user_name").isEqual("Mary"));
        /**
         * 5.2 将table 对象 转为类似mysql视图
         */
        tableEnv.createTemporaryView("resultTable", resultTable);
        /**
         * 5.3 可以基于视图的操作 直接查询
         */
        Table selectResult = tableEnv.sqlQuery("select url, user_name from resultTable");

        /**
         * 5.4 执行聚合计算的查询转换
         */
        Table aggResult = tableEnv.sqlQuery("select user_name, count(url) as cnt  from " + tableName + " group by user_name");
        //   Table aggResult =  clickTable.groupBy($("user_name")).select($("user_name")).select($("COUNT(url) as cnt"));


        /**
         * 6.1 创建输出 虚拟表
         */
        String createOutDdl = "CREATE TABLE outTable (" +
                " url STRING, " +
                " user_name STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'output', " +
                " 'format' =  'csv' " +
                ")";
        tableEnv.executeSql(createOutDdl);

        /**
         * 6.2 创建一张用于控制台打印输出的表
         */
        String createPrintOutDdl = "CREATE TABLE printOutTable (" +
                " user_name STRING, " +
                " cnt BIGINT " +
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";
        tableEnv.executeSql(createPrintOutDdl);

        /**
         * 7 将表数据 输出 字段需要和输出表对应
         */
        aggResult.executeInsert("printOutTable");
        selectResult.executeInsert("outTable");


    }
}
